package DianShang_2024.ds_01.indicator

import org.apache.spark.sql.{DataFrame, SparkSession}

import java.util.Properties

object indicator03 {
  def main(args: Array[String]): Unit = {
    /*
          请根据dwd层表计算出2020年4月每个省份的平均订单金额和所有省份平均订单金额相比较结果（“高/低/相同”）,存入MySQL数据库shtd_result的provinceavgcmp表
          （表结构如下）中，然后在Linux的MySQL命令行中根据省份表主键、该省平均订单金额均为降序排序，查询出前5条，将SQL语句复制粘贴至客户端桌面【Release\任务B提交
          结果.docx】中对应的任务序号下，
     */

    //  准备sparksql的环境
    val spark=SparkSession.builder()
      .appName("指标计算第三题")
      .master("local[*]")
      .enableHiveSupport()
      .getOrCreate()

    spark.sql("use dwd")
    //  准备连接mysql的配置
    val jdbc_conf=new Properties()
    jdbc_conf.setProperty("user","root")
    jdbc_conf.setProperty("password","123456")
    jdbc_conf.setProperty("driver","com.mysql.jdbc.Driver")

    /*
          1.首先查询出所有省份的平均订单金额
          2.然后得到4月份每个省份的平均订单金额
          3.最后进行判断每个省份是否高于全国的订单金额

          and year(t1.create_time)=2020 and month(t1.create_time)=4     :连接的限制条件,表示只取2020年4月份的
          if(判断,结果1,if())                             :嵌套判断
          下面还有一个地方,是将查询当作了一个字段,然后就是在外层才给表另起名字，在内层是不可以使用的，否则会报错
     */

    val result:DataFrame=spark.sql(
      """
        |select
        |province_id,
        |province_name,
        |datas_avg,
        |alldatas_avg,
        |if(
        |datas_avg > alldatas_avg,'高',if(datas_avg < alldatas_avg ,'低','相同')
        |)  as  result
        |from(
        |select
        |t1.province_id as province_id,
        |t2.name as province_name,
        |avg(t1.final_total_amount) as datas_avg,
        |(
        |select
        |avg(final_total_amount)
        |from dwd.fact_order_info
        |where etl_date='20231017'
        |) as alldatas_avg
        |from  dwd.fact_order_info as t1
        |join dwd.dim_province as t2
        |on t1.province_id=t2.id and t1.etl_date='20231017'  and t2.etl_date='20231017'
        |and year(t1.create_time)=2020 and month(t1.create_time)=4
        |group by province_id,province_name
        |) as result_table
        |""".stripMargin)


    //  如果使用的是 overwrite的话那么mysql里面都不需要创建表，也可以直接将数据转输入到mysql对应的数据库下面
    result.write
      .mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110/shtd_result?useSSL=false","province_avgcmp",jdbc_conf)



    //  关闭sparksql的环境
    spark.close()
  }

}
